package cn.ec.service;

import cn.ec.utils.CamelUtil;
import cn.ec.utils.RedisUtil;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.protobuf.InvalidProtocolBufferException;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * canal服务
 * 接收并处理binLog日志
 */
@Slf4j
public class CanalService {

    private static final int batchSize = 100;

    /**
     * 订阅数据库表
     * 0. 连接数据库
     * 1. 订阅
     * 2. 从binLog获取数据
     * 3. 处理binLog的数据
     */
    public static void subscribe() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress("lucky-bin.cn",
                11111), "example", "canal", "canal");
        try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(batchSize);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                    try {
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                } else {
                    doEntries(message.getEntries());
                }
                connector.ack(batchId); // 提交确认
                // connector.rollback(batchId); // 处理失败, 回滚数据
            }
        } finally {
            connector.disconnect();
        }
    }

    /**
     * 记录日志
     *
     * @param entry entry
     * @param eventType 事件类型
     */
    private static void printBinLog(CanalEntry.Entry entry, CanalEntry.EventType eventType) {
        log.debug("binlog[{}:{}],name[{},{}],eventType[{}]",
                entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType);
    }

    /**
     * 处理binLog数据
     * 0. 如果类型为事务则放行
     * 1. 获取发生改变的数据
     * 2. 根据数据库操作(INSERT,UPDATE,DELETE)进行处理
     *
     * @param entries x
     */
    private static void doEntries(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (InvalidProtocolBufferException e) {
                log.error("解析binLog数据失败[{}]", e.getMessage());
                continue;
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            // 打印binLog日志
            printBinLog(entry, eventType);
            // 遍历rowData,根据操作类型进行处理
            final String tableName = entry.getHeader().getTableName();
            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                doRowData(tableName, rowData, eventType);
            }
        }
    }

    /**
     * 处理rowData
     */
    private static void doRowData(String tableName, CanalEntry.RowData rowData, CanalEntry.EventType eventType) {
        if (eventType == CanalEntry.EventType.DELETE) {
            List<CanalEntry.Column> beforeColumnsList = rowData.getBeforeColumnsList();
            log.debug("beforeColumnsList[{}]", beforeColumnsList);
            redisDelete(tableName, beforeColumnsList);
        } else {
            List<CanalEntry.Column> afterColumnsList = rowData.getAfterColumnsList();
            if (eventType == CanalEntry.EventType.INSERT) {
                redisInsert(tableName, afterColumnsList);
            } else {
                log.debug("afterColumnsList[{}]", afterColumnsList);
                redisUpdate(tableName, afterColumnsList);
            }
        }
    }

    private static void redisInsert(String tableName, List<CanalEntry.Column> columns) {
        Map<String, Object> map = new HashMap<>(columns.size());
        for (CanalEntry.Column column : columns) {
            map.put(CamelUtil.toCamel(column.getName()), column.getValue());
        }
        if (columns.size() > 0) {
            ObjectMapper mapper = new ObjectMapper();
            String s = null;
            try {
                s = mapper.writeValueAsString(map);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            log.debug("[INSERT]ID[{}][{}]", columns.get(0).getValue(), s);
            RedisUtil.hashSet(tableName, columns.get(0).getValue(), s);
        }
    }

    private static void redisUpdate(String tableName, List<CanalEntry.Column> columns) {
        Map<String, Object> map = new HashMap<>(columns.size());
        for (CanalEntry.Column column : columns) {
            log.info("[{}][{}]", CamelUtil.toCamel(column.getName()), column.getValue());
            map.put(CamelUtil.toCamel(column.getName()), column.getValue());
        }
        if (columns.size() > 0) {
            ObjectMapper mapper = new ObjectMapper();
            String s = null;
            try {
                s = mapper.writeValueAsString(map);
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
            log.debug("[UPDATE]ID[{}][{}]", columns.get(0).getValue(), s);
            RedisUtil.hashSet(tableName, columns.get(0).getValue(), s);
        }
    }

    private static void redisDelete(String tableName, List<CanalEntry.Column> columns) {
        if (columns.size() > 0) {
            log.debug("[DELETE]ID[{}]", columns.get(0).getValue());
            RedisUtil.hashDel(tableName, columns.get(0).getValue());
        }
    }


}
